// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: jshin

// https://github.com/grpc/grpc/blob/master/examples/cpp/helloworld/greeter_client.cc

#include "coin2/strategy/util/config_reporter.h"

#include <chrono>
#include <vector>

#include <glog/logging.h>
#include <grpcpp/grpcpp.h>

#include "coin2/strategy/config_service.grpc.pb.h"

inline int64_t GetCurrentTimestamp() {
  return std::chrono::duration_cast<std::chrono::nanoseconds>(
             std::chrono::system_clock::now().time_since_epoch())
      .count();
}

template <typename T>
std::string StatusToString(const T& status) {
  std::stringstream ss;
  ss << "error_code: " << status.error_code() << "\n";
  ss << "error_details: " << status.error_details() << "\n";
  ss << "error_message: " << status.error_message() << "\n";
  return ss.str();
}

namespace coin2::strategy::util {

struct ConfigReporter_GrpcImpl {
  explicit ConfigReporter_GrpcImpl(const std::string& server_address) {
    channel = grpc::CreateChannel(server_address, grpc::InsecureChannelCredentials());
    stub = ConfigService::NewStub(channel);
  }

  std::shared_ptr<grpc::Channel> channel;
  std::unique_ptr<ConfigService::StubInterface> stub;
};

ConfigReporter::ConfigReporter(
    const std::string& strategy_name,
    const std::string& server_address,
    const std::vector<std::string>& symbols,
    Handler handler)
    : strategy_name_(strategy_name),
      server_address_(server_address),
      symbols_(symbols),
      handler_(handler) {}

ConfigReporter::~ConfigReporter() {
  if (running_) {
    Shutdown();
  }
}

void ConfigReporter::CreateConnection() {
  grpc_impl_.reset(new ConfigReporter_GrpcImpl(server_address_));
}

void ConfigReporter::UploadConfig() {
  std::string buf;
  for (const auto &prod : symbols_) {
    SetConfigRequest set_req;
    set_req.set_strategy_name(strategy_name_);
    set_req.set_type(ConfigType::TICKER_INFO);
    set_req.set_symbol(prod);
    set_req.SerializeToString(&buf);

    InternalControlRequest ctrl_req;
    ctrl_req.set_action(InternalControlRequest::GET);
    ctrl_req.set_target("/strategy/ticker_info");
    ctrl_req.set_timestamp(GetCurrentTimestamp());
    ctrl_req.set_body(buf);
    SubmitInternalControlRequest(ctrl_req);
    set_req.mutable_ticker_info()->ParseFromString(ctrl_resp_.body());
    set_req.set_config_timestamp(ctrl_req.timestamp());

    LOG(INFO) << "SetConfigRequest: " << set_req.DebugString();
    if (ctrl_resp_.error_code() == 0 && is_grpc_ready()) {
      grpc::ClientContext context;
      SetConfigResponse set_res;
      auto status = grpc_impl_->stub->SetConfig(&context, set_req, &set_res);
      if (status.ok()) {
        LOG(INFO) << set_res.DebugString();
      } else {
        LOG(ERROR) << "grpc Failed to upload ticker info." << StatusToString(status);
      }
    }
  }
}

void ConfigReporter::DownloadConfig() {
  for (const auto &prod : symbols_) {
    GetConfigRequest get_req;
    GetConfigResponse get_res;

    get_req.set_strategy_name(strategy_name_);
    get_req.set_type(ConfigType::RMM_CONFIG);
    get_req.set_symbol(prod);

    if (is_grpc_ready()) {
      grpc::ClientContext context;
      auto status = grpc_impl_->stub->GetConfig(&context, get_req, &get_res);

      if (status.ok()) {
        if (get_res.has_rmm_config()) {
          std::string buf;
          CHECK(get_res.SerializeToString(&buf));
          InternalControlRequest ctrl_req;
          ctrl_req.set_action(InternalControlRequest::SET);
          ctrl_req.set_body(buf);
          SubmitInternalControlRequest(ctrl_req);

          if (ctrl_resp_.error_code() == 0) {
            LOG(INFO) << "rmm control config applied: " << get_res.DebugString();
          } else {
            LOG(ERROR) << "rmm control config failed: " << get_res.DebugString();
          }
        } else {
          LOG(ERROR) << "rmm control config not found in response: " << get_res.DebugString();
        }
      } else {
        LOG(ERROR) << "grpc query failed for rmm config" << StatusToString(status);
      }
    }
  }

  for (const auto &prod : symbols_) {
    GetConfigRequest get_req;
    GetConfigResponse get_res;

    get_req.set_strategy_name(strategy_name_);
    get_req.set_type(ConfigType::SCALE_INFO);
    get_req.set_symbol(prod);

    if (is_grpc_ready()) {
      grpc::ClientContext context;
      auto status = grpc_impl_->stub->GetConfig(&context, get_req, &get_res);
      if (status.ok()) {
        if (get_res.has_scale_info()) {
          std::string buf;
          CHECK(get_res.SerializeToString(&buf));
          InternalControlRequest ctrl_req;
          ctrl_req.set_action(InternalControlRequest::SET);
          ctrl_req.set_body(buf);
          SubmitInternalControlRequest(ctrl_req);

          if (ctrl_resp_.error_code() == 0) {
            LOG(INFO) << "scale adjusted!: " << get_res.DebugString();
          } else {
            LOG(ERROR) << "scale adjusted failed: " << get_res.DebugString();
          }
        } else {
          LOG(ERROR) << "scale adjusted missing from return!" << get_res.DebugString();
        }
      } else {
        LOG(ERROR) << "grpc query failed for scaleinfo" << StatusToString(status);
      }
    }
  }
}

void ConfigReporter::Loop() {
  using namespace std::chrono_literals;

  CreateConnection();
  while (running_) {
    UploadConfig();
    DownloadConfig();
    std::this_thread::sleep_for(5s);
  }
}

bool ConfigReporter::is_grpc_ready() const {
  // check grpc ready
  return static_cast<bool>(grpc_impl_ && grpc_impl_->stub);
}

void ConfigReporter::SubmitInternalControlRequest(const InternalControlRequest ctrl_req) {
  ctrl_resp_.Clear();
  ctrl_resp_.set_error_code(1);
  ctrl_resp_.set_error_msg("NO_RESPONSE");
  ctrl_resp_.mutable_request()->CopyFrom(ctrl_req);

  if (handler_) {
    handler_(ctrl_req);
  }
}

void ConfigReporter::Start() {
  running_ = true;
  thread_ = std::thread([this]() { this->Loop(); });
  LOG(INFO) << "ConfigReporter started!";
}

void ConfigReporter::Shutdown() {
  running_ = false;
  thread_.join();
}

void ConfigReporter::SetInternalControlResponse(const InternalControlResponse ctrl_resp) {
  ctrl_resp_.CopyFrom(ctrl_resp);
}

}  // namespace coin2::strategy::util
